package com.uestc.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.NullArgumentException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.jasper.JasperException;

public class MTable {

	private Table mTable;
	private Result mResult;

	/**
	 * 枚举类型，用于枚举过滤器的种类
	 * 
	 * @author lijiong
	 *
	 */
	public static enum FILTER_TYPE {
		PREFIX_FILTER("PREFIX_FILTER"), SINGLE_COLUMN_VALUE_FILTER("SINGLE_COLUMN_VALUE_FILTER"), ROW_FILTER(
				"ROW_FILTER");
		private String filterName;

		private FILTER_TYPE(String s) {
			filterName = s;
		}

		public String getFilterName() {
			return filterName;
		}

		@Override
		public String toString() {
			return filterName;
		}
	};

	/**
	 * 各个过滤器的结合方式
	 * 
	 * @author lijiong
	 *
	 */
	public static enum FILTER_COMBINE {
		MUST_PASS_ALL("must_past_all"), MUST_PASS_ONE("must_past_one"), INCLUDE("include"), EQUAL("equal");
		private String name;

		private FILTER_COMBINE(String s) {
			name = s;
		}

		@Override
		public String toString() {
			return name;
		}
	}

	public void test() throws JasperException {
		throw new JasperException("");
	}

	/**
	 * 构造函数
	 * 
	 * @param tableName
	 *            利用表名和行键来进行构造
	 * @throws IOException
	 */
	public MTable(String tableName) throws IOException {
		if (tableName == null || tableName.equals(""))
			throw new NullArgumentException("tableName是null或空字符串");
		mTable = HBaseUtil.getHTable(tableName);
	}

	/**
	 * 构造函数，利用表名和行键进行构造
	 * 
	 * @param tableName
	 * @param row
	 * @throws IOException
	 */
	public MTable(String tableName, String row) throws IOException {
		mTable = HBaseUtil.getHTable(tableName);
		mResult = mTable.get(new Get(Bytes.toBytes(row)));
	}

	public void getRowInfo(String row) throws IOException {
		if (mTable == null)
			throw new NullPointerException();
		mResult = mTable.get(new Get(Bytes.toBytes(row)));
	}

	public Table getTable() {
		return mTable;
	}

	/**
	 * 根据行名、列族名和列名来获取对应的值
	 * 
	 * @param row
	 *            行名
	 * @param family
	 *            列族
	 * @param qualifier
	 *            列
	 * @return
	 * @throws IOException
	 */
	public String getValue(String row, String family, String qualifier) throws IOException {
		if (isArgumentNullOrEmpty(row, family, qualifier)) {
			throw new NullArgumentException("row、family或qualifier存在null或空字符串");
		}
		Get get = new Get(Bytes.toBytes(row));
		Result result = mTable.get(get);
		byte[] bytes = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
		return Bytes.toString(bytes);
	}

	/**
	 * 在Result结果集已经获得的情况下，直接用列族名和列名来获取对应的值
	 * 
	 * @param family
	 *            列族
	 * @param qualifier
	 *            列族下的列名
	 * @return
	 * @throws IOException
	 */
	public String getValue(String family, String qualifier) throws IOException {
		if (mResult == null)
			throw new NullArgumentException("mResult参数是null");

		if (isArgumentNullOrEmpty(family, qualifier)) {
			throw new NullArgumentException("row、family或qualifier存在null或空字符串");
		}

		byte[] bytes = mResult.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
		return Bytes.toString(bytes);
	}

	public LinkedHashMap<byte[], Map<String, Map<String, String>>> getPage(int pageSize, byte[] row,
			List<String> queryList) throws IOException {
		return this.getPage(pageSize, row, queryList, null, null, null, null);
	}

	/**
	 * 在HTable里进行分页查找，返回结果有一点复杂, 添加的两个条件是过滤
	 * 
	 * @param pageSize
	 *            每页的数据量
	 * @param row
	 *            行键
	 * @param queryList
	 *            要查询的列族组成的List
	 * @return 返回一个LinkedHashMap,它的key是行键，类型是byte[]，这个LinkedHashMap是按插入的顺序排列的，因此
	 *         ，它的最后一行可以缓存起来，用于下一次的起始行。LinkedHashMap的value又是一个Map，
	 *         这个Map的Key是列族的名称，它的value还是一个Map，这个Map的key是qualifier，value是列值
	 * 
	 * @throws IOException
	 * 
	 */
	public LinkedHashMap<byte[], Map<String, Map<String, String>>> getPage(int pageSize, byte[] row,
			List<String> queryList, String fam, HashMap<String, String> conditions, FILTER_TYPE filtertype,
			FILTER_COMBINE conditionsFilter) throws IOException {
		if (pageSize < 1 || queryList == null)
			throw new NullArgumentException("argument is illegal");
		byte[] startRow = row; // 开始的行
		Filter filter = null;
		Scan scan = new Scan();
		if (fam == null || conditions == null)
			filter = new PageFilter(pageSize + 1); // 这里多处理一行，主要是为了获得最后一行的信息，便于翻页时用于startRow
		else {
			ArrayList<Filter> filters = new ArrayList<>();
			// 设置过滤器
			for (Map.Entry<String, String> entry : conditions.entrySet()) {
				String qualifier = entry.getKey();
				String value = entry.getValue();
				if (FILTER_TYPE.PREFIX_FILTER == filtertype || "userRowKey".equals(qualifier)) {
					filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(value));
					// scan.setRowPrefixFilter(Bytes.toBytes(value));
				} else if (FILTER_TYPE.SINGLE_COLUMN_VALUE_FILTER == filtertype)
					filter = new SingleColumnValueFilter(Bytes.toBytes(fam), Bytes.toBytes(qualifier),
							CompareFilter.CompareOp.EQUAL, Bytes.toBytes(value));
				if (filter != null)
					filters.add(filter);
			}
			if (FILTER_COMBINE.MUST_PASS_ALL == conditionsFilter) {
				filter = new PageFilter(pageSize + 1);
				filters.add(filter);
				filter = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);
			} else {
				filter = new FilterList(FilterList.Operator.MUST_PASS_ONE, filters);
				Filter pageFilter = new PageFilter(pageSize + 1);
				filters.clear();
				filters.add(filter);
				filters.add(pageFilter);
				filter = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);
			}
			scan.addFamily(Bytes.toBytes(fam)); // 设置把这个列族的信息全部scan出来
		}
		scan.setSmall(true); // 扫描少量数据，根据源码注释设置这个值能提高性能
		if (startRow != null) {
			scan.setStartRow(startRow);
		}
		scan.setFilter(filter);
		ResultScanner scanner = mTable.getScanner(scan);
		// resultMap存放行键即key，以及行键对应的许多列族即value，value是由列族名和列族对应的qualifier以及value组成
		LinkedHashMap<byte[], Map<String, Map<String, String>>> resultMap = new LinkedHashMap<>();
		for (Result result : scanner) {
			Map<String, Map<String, String>> familyMap = new HashMap<>();
			for (String family : queryList) {
				Map<byte[], byte[]> tempQualifierMap = result.getFamilyMap(Bytes.toBytes(family));
				Map<String, String> qualifierMap = new HashMap<>();
				// 循环改变Map的类型
				for (Map.Entry<byte[], byte[]> item : tempQualifierMap.entrySet()) {
					qualifierMap.put(Bytes.toString(item.getKey()), Bytes.toString(item.getValue()));
				}
				familyMap.put(family, qualifierMap);
			}
			resultMap.put(result.getRow(), familyMap);
		}
		scanner.close();
		return resultMap;
	}

	/**
	 * 统计一个Table的行数，需要给出表名、列族名和列名 需要在hbase中添加Coprocessor配置
	 * <property> <name>hbase.coprocessor.region.classes</name>
	 * <value>org.apache.hadoop.hbase.coprocessor.AggregateImplementation
	 * </value> </property>
	 * 
	 * @param tableName
	 * @param family
	 * @param qualifier
	 * @return
	 * @throws Throwable
	 */
	public long getRowCounter(String tableName, String family, String qualifier) throws Throwable {
		if (isArgumentNullOrEmpty(tableName, family, qualifier)) {
			throw new NullArgumentException("argument is null or empty");
		}
		AggregationClient aClient = new AggregationClient(HBaseUtil.getConf());
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
		long rowCount = aClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
		aClient.close();
		return rowCount;
	}

	/**
	 * 带条件查询
	 * 
	 * @param tableName
	 * @param family
	 * @param qualifier
	 * @param condition
	 *            列值
	 * @return
	 * @throws Throwable
	 */
	public long getRowCounterWithCondition(String tableName, String family, String qualifier, String condition)
			throws Throwable {
		return getRowCounterWithCondition(tableName, family, qualifier, condition, FILTER_COMBINE.EQUAL);
	}

	public long getRowCounterWithCondition(String tableName, String family, String qualifier, String condition,
			FILTER_COMBINE fCombine) throws Throwable {
		if (isArgumentNullOrEmpty(tableName, family, qualifier, condition)) {
			throw new NullArgumentException("argument is null or empty");
		}
		SingleColumnValueFilter filter = null;
		if (fCombine == FILTER_COMBINE.EQUAL)
			filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					CompareFilter.CompareOp.EQUAL, Bytes.toBytes(condition));
		else
			filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					CompareFilter.CompareOp.EQUAL, new SubstringComparator(condition));
		Scan scan = new Scan();
		scan.setFilter(filter);
		scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
		AggregationClient aClient = new AggregationClient(HBaseUtil.getConf());
		long rowCount = aClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
		aClient.close();
		return rowCount;
	}

	public long getRowCounterWithConditions(String tableName, String family, HashMap<String, String> conditions)
			throws Throwable {
		return this.getRowCounterWithConditions(tableName, family, conditions, FILTER_COMBINE.MUST_PASS_ALL);
	}

	public long getRowCounterWithConditions(String tableName, String family, HashMap<String, String> conditions,
			FILTER_COMBINE conditionFilter) throws Throwable {
		if (isArgumentNullOrEmpty(tableName, family) || conditions == null) {
			throw new NullArgumentException("argument is null or empty");
		}
		ArrayList<Filter> filterlist = new ArrayList<>();
		Filter filter = null;
		for (Entry<String, String> item : conditions.entrySet()) {
			String qualifier = item.getKey();
			String value = item.getValue();
			if (qualifier.contains(FILTER_TYPE.ROW_FILTER.toString()) || "userRowKey".equals(qualifier))
				filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(item.getValue()));
			else
				filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier),
						CompareFilter.CompareOp.EQUAL, Bytes.toBytes(value));
			filterlist.add(filter);
		}

		Scan scan = new Scan();
		if (FILTER_COMBINE.MUST_PASS_ALL == conditionFilter) // 判断各个Filter的关系是与还是或
			filter = new FilterList(FilterList.Operator.MUST_PASS_ALL, filterlist);
		else
			filter = new FilterList(FilterList.Operator.MUST_PASS_ONE, filterlist);
		scan.setFilter(filter);
		AggregationClient aClient = new AggregationClient(HBaseUtil.getConf());
		long rowCount = aClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
		aClient.close();
		return rowCount;
	}

	public void close() throws IOException {
		if (mTable != null) {
			mTable.close();
			mTable = null;
		}
	}

	/**
	 * 判断参数是否为null或空，为空则返回true
	 * 
	 * @param arg
	 * @return
	 */
	public boolean isArgumentNullOrEmpty(String... args) {
		for (String arg : args) {
			if (arg == null || arg.equals(""))
				return true;
		}
		return false;
	}

	/**
	 * 给出行键，判断某个行是否存在，这里可直接给出行键，方法会在key前自动加上hash值
	 * 
	 * @param row
	 * @return
	 * @throws IOException
	 */
	public boolean isRowExists(String row) throws IOException {
		Get get = new Get(Bytes.toBytes(row));
		return mTable.exists(get);
	}

	/**
	 * 添加单个数据
	 * 
	 * @param row
	 * @param family
	 * @param qualifier
	 * @param value
	 * @throws IOException
	 */
	public void setData(String row, String family, String qualifier, String value) throws IOException {
		Put put = new Put(Bytes.toBytes(row));
		put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
		mTable.put(put);
	}

	/**
	 * 添加多个数据
	 * 
	 * @param row
	 * @param family
	 * @param items
	 * @throws IOException
	 */
	public void setData(String row, String family, Map<String, String> items) throws IOException {
		if (row == null || row.equals("") || family == null || family.equals("") || items == null || items.size() < 0)
			throw new NullPointerException("argument exist null or ''");
		List<Put> putList = new ArrayList<Put>();
		for (Map.Entry<String, String> item : items.entrySet()) {
			String qualifier = item.getKey();
			String value = item.getValue();
			Put put = new Put(Bytes.toBytes(row));
			put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
			putList.add(put);
		}
		mTable.put(putList);
	}

	public ResultScanner getScanner(Scan scan) throws IOException {
		return mTable.getScanner(scan);
	}
}
